/*
 * Licensed to the Apache Software Foundation (ASF) under one or more
 * contributor license agreements.  See the NOTICE file distributed with
 * this work for additional information regarding copyright ownership.
 * The ASF licenses this file to You under the Apache License, Version 2.0
 * (the "License"); you may not use this file except in compliance with
 * the License.  You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package org.apache.rocketmq.namesrv.processor;

import io.netty.channel.ChannelHandlerContext;
import org.apache.rocketmq.common.DataVersion;
import org.apache.rocketmq.common.MQVersion;
import org.apache.rocketmq.common.MQVersion.Version;
import org.apache.rocketmq.common.MixAll;
import org.apache.rocketmq.common.UtilAll;
import org.apache.rocketmq.common.constant.LoggerName;
import org.apache.rocketmq.common.help.FAQUrl;
import org.apache.rocketmq.common.namesrv.NamesrvUtil;
import org.apache.rocketmq.common.namesrv.RegisterBrokerResult;
import org.apache.rocketmq.common.protocol.RequestCode;
import org.apache.rocketmq.common.protocol.ResponseCode;
import org.apache.rocketmq.common.protocol.body.RegisterBrokerBody;
import org.apache.rocketmq.common.protocol.body.TopicConfigSerializeWrapper;
import org.apache.rocketmq.common.protocol.header.GetTopicsByClusterRequestHeader;
import org.apache.rocketmq.common.protocol.header.namesrv.*;
import org.apache.rocketmq.common.protocol.route.TopicRouteData;
import org.apache.rocketmq.logging.InternalLogger;
import org.apache.rocketmq.logging.InternalLoggerFactory;
import org.apache.rocketmq.namesrv.NamesrvController;
import org.apache.rocketmq.remoting.common.RemotingHelper;
import org.apache.rocketmq.remoting.exception.RemotingCommandException;
import org.apache.rocketmq.remoting.netty.NettyRequestProcessor;
import org.apache.rocketmq.remoting.protocol.RemotingCommand;

import java.io.UnsupportedEncodingException;
import java.util.Properties;
import java.util.concurrent.atomic.AtomicLong;

/** NameServer处理请求的请求类，负责处理所有与NameServer交互的请求； */
public class DefaultRequestProcessor implements NettyRequestProcessor {
  private static InternalLogger log =
      InternalLoggerFactory.getLogger(LoggerName.NAMESRV_LOGGER_NAME);

  protected final NamesrvController namesrvController;

  public DefaultRequestProcessor(NamesrvController namesrvController) {
    this.namesrvController = namesrvController;
  }

  @Override
  public RemotingCommand processRequest(ChannelHandlerContext ctx, RemotingCommand request)
      throws RemotingCommandException {

    if (ctx != null) {
      log.debug(
          "receive request, {} {} {}",
          request.getCode(),
          RemotingHelper.parseChannelRemoteAddr(ctx.channel()),
          request);
    }

    // 解析请求类型
    switch (request.getCode()) {
      case RequestCode.PUT_KV_CONFIG:
        return this.putKVConfig(ctx, request);
      case RequestCode.GET_KV_CONFIG:
        return this.getKVConfig(ctx, request);
      case RequestCode.DELETE_KV_CONFIG:
        return this.deleteKVConfig(ctx, request);
      case RequestCode.QUERY_DATA_VERSION:
        return queryBrokerTopicConfig(ctx, request);
        // 如果是RequestCode.REGISTER_BROKER，则请求最终转发到routeManager 的 regisbroker
      case RequestCode.REGISTER_BROKER:
        Version brokerVersion = MQVersion.value2Version(request.getVersion());
        // 如果broker的版本大于或则等于V3_0_11
        if (brokerVersion.ordinal() >= MQVersion.Version.V3_0_11.ordinal()) {
          return this.registerBrokerWithFilterServer(ctx, request);
        } else {
          return this.registerBroker(ctx, request);
        }
      case RequestCode.UNREGISTER_BROKER:
        return this.unregisterBroker(ctx, request);
        // 如果命令编码为GET_ROUTEINTO_BY_TOPIC
      case RequestCode.GET_ROUTEINTO_BY_TOPIC:
        return this.getRouteInfoByTopic(ctx, request);
      case RequestCode.GET_BROKER_CLUSTER_INFO:
        return this.getBrokerClusterInfo(ctx, request);
      case RequestCode.WIPE_WRITE_PERM_OF_BROKER:
        return this.wipeWritePermOfBroker(ctx, request);
      case RequestCode.GET_ALL_TOPIC_LIST_FROM_NAMESERVER:
        return getAllTopicListFromNameserver(ctx, request);
      case RequestCode.DELETE_TOPIC_IN_NAMESRV:
        return deleteTopicInNamesrv(ctx, request);
      case RequestCode.GET_KVLIST_BY_NAMESPACE:
        return this.getKVListByNamespace(ctx, request);
      case RequestCode.GET_TOPICS_BY_CLUSTER:
        return this.getTopicsByCluster(ctx, request);
      case RequestCode.GET_SYSTEM_TOPIC_LIST_FROM_NS:
        return this.getSystemTopicListFromNs(ctx, request);
      case RequestCode.GET_UNIT_TOPIC_LIST:
        return this.getUnitTopicList(ctx, request);
      case RequestCode.GET_HAS_UNIT_SUB_TOPIC_LIST:
        return this.getHasUnitSubTopicList(ctx, request);
      case RequestCode.GET_HAS_UNIT_SUB_UNUNIT_TOPIC_LIST:
        return this.getHasUnitSubUnUnitTopicList(ctx, request);
      case RequestCode.UPDATE_NAMESRV_CONFIG:
        return this.updateConfig(ctx, request);
      case RequestCode.GET_NAMESRV_CONFIG:
        return this.getConfig(ctx, request);
      default:
        break;
    }
    return null;
  }

  @Override
  public boolean rejectRequest() {
    return false;
  }

  public RemotingCommand putKVConfig(ChannelHandlerContext ctx, RemotingCommand request)
      throws RemotingCommandException {
    final RemotingCommand response = RemotingCommand.createResponseCommand(null);
    final PutKVConfigRequestHeader requestHeader =
        (PutKVConfigRequestHeader)
            request.decodeCommandCustomHeader(PutKVConfigRequestHeader.class);

    this.namesrvController
        .getKvConfigManager()
        .putKVConfig(
            requestHeader.getNamespace(), requestHeader.getKey(), requestHeader.getValue());

    response.setCode(ResponseCode.SUCCESS);
    response.setRemark(null);
    return response;
  }

  public RemotingCommand getKVConfig(ChannelHandlerContext ctx, RemotingCommand request)
      throws RemotingCommandException {
    final RemotingCommand response =
        RemotingCommand.createResponseCommand(GetKVConfigResponseHeader.class);
    final GetKVConfigResponseHeader responseHeader =
        (GetKVConfigResponseHeader) response.readCustomHeader();
    final GetKVConfigRequestHeader requestHeader =
        (GetKVConfigRequestHeader)
            request.decodeCommandCustomHeader(GetKVConfigRequestHeader.class);

    String value =
        this.namesrvController
            .getKvConfigManager()
            .getKVConfig(requestHeader.getNamespace(), requestHeader.getKey());

    if (value != null) {
      responseHeader.setValue(value);
      response.setCode(ResponseCode.SUCCESS);
      response.setRemark(null);
      return response;
    }

    response.setCode(ResponseCode.QUERY_NOT_FOUND);
    response.setRemark(
        "No config item, Namespace: "
            + requestHeader.getNamespace()
            + " Key: "
            + requestHeader.getKey());
    return response;
  }

  public RemotingCommand deleteKVConfig(ChannelHandlerContext ctx, RemotingCommand request)
      throws RemotingCommandException {
    final RemotingCommand response = RemotingCommand.createResponseCommand(null);
    final DeleteKVConfigRequestHeader requestHeader =
        (DeleteKVConfigRequestHeader)
            request.decodeCommandCustomHeader(DeleteKVConfigRequestHeader.class);

    this.namesrvController
        .getKvConfigManager()
        .deleteKVConfig(requestHeader.getNamespace(), requestHeader.getKey());

    response.setCode(ResponseCode.SUCCESS);
    response.setRemark(null);
    return response;
  }

  public RemotingCommand registerBrokerWithFilterServer(
      ChannelHandlerContext ctx, RemotingCommand request) throws RemotingCommandException {
    final RemotingCommand response =
        RemotingCommand.createResponseCommand(RegisterBrokerResponseHeader.class);
    final RegisterBrokerResponseHeader responseHeader =
        (RegisterBrokerResponseHeader) response.readCustomHeader();
    final RegisterBrokerRequestHeader requestHeader =
        (RegisterBrokerRequestHeader)
            request.decodeCommandCustomHeader(RegisterBrokerRequestHeader.class);

    if (!checksum(ctx, request, requestHeader)) {
      response.setCode(ResponseCode.SYSTEM_ERROR);
      response.setRemark("crc32 not match");
      return response;
    }

    RegisterBrokerBody registerBrokerBody = new RegisterBrokerBody();

    if (request.getBody() != null) {
      try {
        registerBrokerBody =
            RegisterBrokerBody.decode(request.getBody(), requestHeader.isCompressed());
      } catch (Exception e) {
        throw new RemotingCommandException("Failed to decode RegisterBrokerBody", e);
      }
    } else {
      registerBrokerBody
          .getTopicConfigSerializeWrapper()
          .getDataVersion()
          .setCounter(new AtomicLong(0));
      registerBrokerBody.getTopicConfigSerializeWrapper().getDataVersion().setTimestamp(0);
    }

    // 从RegisterBrokerRequestHeader获取到broker传递过来的信息，调用路由元数据的注册broker接口。
    // 返回一个注册的结果
    RegisterBrokerResult result =
        this.namesrvController
            .getRouteInfoManager()
            .registerBroker(
                requestHeader.getClusterName(),
                requestHeader.getBrokerAddr(),
                requestHeader.getBrokerName(),
                requestHeader.getBrokerId(),
                requestHeader.getHaServerAddr(),
                registerBrokerBody.getTopicConfigSerializeWrapper(),
                registerBrokerBody.getFilterServerList(),
                ctx.channel());

    responseHeader.setHaServerAddr(result.getHaServerAddr());
    responseHeader.setMasterAddr(result.getMasterAddr());

    byte[] jsonValue =
        this.namesrvController
            .getKvConfigManager()
            .getKVListByNamespace(NamesrvUtil.NAMESPACE_ORDER_TOPIC_CONFIG);
    response.setBody(jsonValue);

    response.setCode(ResponseCode.SUCCESS);
    response.setRemark(null);
    return response;
  }

  private boolean checksum(
      ChannelHandlerContext ctx,
      RemotingCommand request,
      RegisterBrokerRequestHeader requestHeader) {
    if (requestHeader.getBodyCrc32() != 0) {
      final int crc32 = UtilAll.crc32(request.getBody());
      if (crc32 != requestHeader.getBodyCrc32()) {
        log.warn(
            String.format(
                "receive registerBroker request,crc32 not match,from %s",
                RemotingHelper.parseChannelRemoteAddr(ctx.channel())));
        return false;
      }
    }
    return true;
  }

  public RemotingCommand queryBrokerTopicConfig(ChannelHandlerContext ctx, RemotingCommand request)
      throws RemotingCommandException {
    final RemotingCommand response =
        RemotingCommand.createResponseCommand(QueryDataVersionResponseHeader.class);
    final QueryDataVersionResponseHeader responseHeader =
        (QueryDataVersionResponseHeader) response.readCustomHeader();
    final QueryDataVersionRequestHeader requestHeader =
        (QueryDataVersionRequestHeader)
            request.decodeCommandCustomHeader(QueryDataVersionRequestHeader.class);
    DataVersion dataVersion = DataVersion.decode(request.getBody(), DataVersion.class);

    Boolean changed =
        this.namesrvController
            .getRouteInfoManager()
            .isBrokerTopicConfigChanged(requestHeader.getBrokerAddr(), dataVersion);
    if (!changed) {
      this.namesrvController
          .getRouteInfoManager()
          .updateBrokerInfoUpdateTimestamp(requestHeader.getBrokerAddr());
    }

    DataVersion nameSeverDataVersion =
        this.namesrvController
            .getRouteInfoManager()
            .queryBrokerTopicConfig(requestHeader.getBrokerAddr());
    response.setCode(ResponseCode.SUCCESS);
    response.setRemark(null);

    if (nameSeverDataVersion != null) {
      response.setBody(nameSeverDataVersion.encode());
    }
    responseHeader.setChanged(changed);
    return response;
  }

  public RemotingCommand registerBroker(ChannelHandlerContext ctx, RemotingCommand request)
      throws RemotingCommandException {
    final RemotingCommand response =
        RemotingCommand.createResponseCommand(RegisterBrokerResponseHeader.class);
    final RegisterBrokerResponseHeader responseHeader =
        (RegisterBrokerResponseHeader) response.readCustomHeader();
    final RegisterBrokerRequestHeader requestHeader =
        (RegisterBrokerRequestHeader)
            request.decodeCommandCustomHeader(RegisterBrokerRequestHeader.class);

    if (!checksum(ctx, request, requestHeader)) {
      response.setCode(ResponseCode.SYSTEM_ERROR);
      response.setRemark("crc32 not match");
      return response;
    }

    TopicConfigSerializeWrapper topicConfigWrapper;
    if (request.getBody() != null) {
      topicConfigWrapper =
          TopicConfigSerializeWrapper.decode(request.getBody(), TopicConfigSerializeWrapper.class);
    } else {
      topicConfigWrapper = new TopicConfigSerializeWrapper();
      topicConfigWrapper.getDataVersion().setCounter(new AtomicLong(0));
      topicConfigWrapper.getDataVersion().setTimestamp(0);
    }

    RegisterBrokerResult result =
        this.namesrvController
            .getRouteInfoManager()
            .registerBroker(
                requestHeader.getClusterName(),
                requestHeader.getBrokerAddr(),
                requestHeader.getBrokerName(),
                requestHeader.getBrokerId(),
                requestHeader.getHaServerAddr(),
                topicConfigWrapper,
                null,
                ctx.channel());

    responseHeader.setHaServerAddr(result.getHaServerAddr());
    responseHeader.setMasterAddr(result.getMasterAddr());

    byte[] jsonValue =
        this.namesrvController
            .getKvConfigManager()
            .getKVListByNamespace(NamesrvUtil.NAMESPACE_ORDER_TOPIC_CONFIG);
    response.setBody(jsonValue);
    response.setCode(ResponseCode.SUCCESS);
    response.setRemark(null);
    return response;
  }

  public RemotingCommand unregisterBroker(ChannelHandlerContext ctx, RemotingCommand request)
      throws RemotingCommandException {
    final RemotingCommand response = RemotingCommand.createResponseCommand(null);
    final UnRegisterBrokerRequestHeader requestHeader =
        (UnRegisterBrokerRequestHeader)
            request.decodeCommandCustomHeader(UnRegisterBrokerRequestHeader.class);

    this.namesrvController
        .getRouteInfoManager()
        .unregisterBroker(
            requestHeader.getClusterName(),
            requestHeader.getBrokerAddr(),
            requestHeader.getBrokerName(),
            requestHeader.getBrokerId());

    response.setCode(ResponseCode.SUCCESS);
    response.setRemark(null);
    return response;
  }

  public RemotingCommand getRouteInfoByTopic(ChannelHandlerContext ctx, RemotingCommand request)
      throws RemotingCommandException {
    final RemotingCommand response = RemotingCommand.createResponseCommand(null);
    final GetRouteInfoRequestHeader requestHeader =
        (GetRouteInfoRequestHeader)
            request.decodeCommandCustomHeader(GetRouteInfoRequestHeader.class);

    TopicRouteData topicRouteData =
        this.namesrvController.getRouteInfoManager().pickupTopicRouteData(requestHeader.getTopic());

    // 如果找到主题
    if (topicRouteData != null) {
      // 并且该主题为【顺序消息】
      if (this.namesrvController.getNamesrvConfig().isOrderMessageEnable()) {
        String orderTopicConf =
            this.namesrvController
                .getKvConfigManager()
                .getKVConfig(NamesrvUtil.NAMESPACE_ORDER_TOPIC_CONFIG, requestHeader.getTopic());
        // 如果是顺序消息，从KVconfig里面去获取顺序消息的配置。设置到路由结果的实体里面。
        topicRouteData.setOrderTopicConf(orderTopicConf);
      }

      // 主题路由编码 放到res 的body里面
      byte[] content = topicRouteData.encode();
      response.setBody(content);
      response.setCode(ResponseCode.SUCCESS);
      response.setRemark(null);
      return response;
    }

    // 没获取到路由
    response.setCode(ResponseCode.TOPIC_NOT_EXIST);
    response.setRemark(
        "No topic route info in name server for the topic: "
            + requestHeader.getTopic()
            + FAQUrl.suggestTodo(FAQUrl.APPLY_TOPIC_URL));
    return response;
  }

  private RemotingCommand getBrokerClusterInfo(ChannelHandlerContext ctx, RemotingCommand request) {
    final RemotingCommand response = RemotingCommand.createResponseCommand(null);

    byte[] content = this.namesrvController.getRouteInfoManager().getAllClusterInfo();
    response.setBody(content);

    response.setCode(ResponseCode.SUCCESS);
    response.setRemark(null);
    return response;
  }

  private RemotingCommand wipeWritePermOfBroker(ChannelHandlerContext ctx, RemotingCommand request)
      throws RemotingCommandException {
    final RemotingCommand response =
        RemotingCommand.createResponseCommand(WipeWritePermOfBrokerResponseHeader.class);
    final WipeWritePermOfBrokerResponseHeader responseHeader =
        (WipeWritePermOfBrokerResponseHeader) response.readCustomHeader();
    final WipeWritePermOfBrokerRequestHeader requestHeader =
        (WipeWritePermOfBrokerRequestHeader)
            request.decodeCommandCustomHeader(WipeWritePermOfBrokerRequestHeader.class);

    int wipeTopicCnt =
        this.namesrvController
            .getRouteInfoManager()
            .wipeWritePermOfBrokerByLock(requestHeader.getBrokerName());

    log.info(
        "wipe write perm of broker[{}], client: {}, {}",
        requestHeader.getBrokerName(),
        RemotingHelper.parseChannelRemoteAddr(ctx.channel()),
        wipeTopicCnt);

    responseHeader.setWipeTopicCount(wipeTopicCnt);
    response.setCode(ResponseCode.SUCCESS);
    response.setRemark(null);
    return response;
  }

  private RemotingCommand getAllTopicListFromNameserver(
      ChannelHandlerContext ctx, RemotingCommand request) {
    final RemotingCommand response = RemotingCommand.createResponseCommand(null);

    byte[] body = this.namesrvController.getRouteInfoManager().getAllTopicList();

    response.setBody(body);
    response.setCode(ResponseCode.SUCCESS);
    response.setRemark(null);
    return response;
  }

  private RemotingCommand deleteTopicInNamesrv(ChannelHandlerContext ctx, RemotingCommand request)
      throws RemotingCommandException {
    final RemotingCommand response = RemotingCommand.createResponseCommand(null);
    final DeleteTopicInNamesrvRequestHeader requestHeader =
        (DeleteTopicInNamesrvRequestHeader)
            request.decodeCommandCustomHeader(DeleteTopicInNamesrvRequestHeader.class);

    this.namesrvController.getRouteInfoManager().deleteTopic(requestHeader.getTopic());

    response.setCode(ResponseCode.SUCCESS);
    response.setRemark(null);
    return response;
  }

  private RemotingCommand getKVListByNamespace(ChannelHandlerContext ctx, RemotingCommand request)
      throws RemotingCommandException {
    final RemotingCommand response = RemotingCommand.createResponseCommand(null);
    final GetKVListByNamespaceRequestHeader requestHeader =
        (GetKVListByNamespaceRequestHeader)
            request.decodeCommandCustomHeader(GetKVListByNamespaceRequestHeader.class);

    byte[] jsonValue =
        this.namesrvController
            .getKvConfigManager()
            .getKVListByNamespace(requestHeader.getNamespace());
    if (null != jsonValue) {
      response.setBody(jsonValue);
      response.setCode(ResponseCode.SUCCESS);
      response.setRemark(null);
      return response;
    }

    response.setCode(ResponseCode.QUERY_NOT_FOUND);
    response.setRemark("No config item, Namespace: " + requestHeader.getNamespace());
    return response;
  }

  private RemotingCommand getTopicsByCluster(ChannelHandlerContext ctx, RemotingCommand request)
      throws RemotingCommandException {
    final RemotingCommand response = RemotingCommand.createResponseCommand(null);
    final GetTopicsByClusterRequestHeader requestHeader =
        (GetTopicsByClusterRequestHeader)
            request.decodeCommandCustomHeader(GetTopicsByClusterRequestHeader.class);

    byte[] body =
        this.namesrvController.getRouteInfoManager().getTopicsByCluster(requestHeader.getCluster());

    response.setBody(body);
    response.setCode(ResponseCode.SUCCESS);
    response.setRemark(null);
    return response;
  }

  private RemotingCommand getSystemTopicListFromNs(
      ChannelHandlerContext ctx, RemotingCommand request) throws RemotingCommandException {
    final RemotingCommand response = RemotingCommand.createResponseCommand(null);

    byte[] body = this.namesrvController.getRouteInfoManager().getSystemTopicList();

    response.setBody(body);
    response.setCode(ResponseCode.SUCCESS);
    response.setRemark(null);
    return response;
  }

  private RemotingCommand getUnitTopicList(ChannelHandlerContext ctx, RemotingCommand request)
      throws RemotingCommandException {
    final RemotingCommand response = RemotingCommand.createResponseCommand(null);

    byte[] body = this.namesrvController.getRouteInfoManager().getUnitTopics();

    response.setBody(body);
    response.setCode(ResponseCode.SUCCESS);
    response.setRemark(null);
    return response;
  }

  private RemotingCommand getHasUnitSubTopicList(ChannelHandlerContext ctx, RemotingCommand request)
      throws RemotingCommandException {
    final RemotingCommand response = RemotingCommand.createResponseCommand(null);

    byte[] body = this.namesrvController.getRouteInfoManager().getHasUnitSubTopicList();

    response.setBody(body);
    response.setCode(ResponseCode.SUCCESS);
    response.setRemark(null);
    return response;
  }

  private RemotingCommand getHasUnitSubUnUnitTopicList(
      ChannelHandlerContext ctx, RemotingCommand request) throws RemotingCommandException {
    final RemotingCommand response = RemotingCommand.createResponseCommand(null);

    byte[] body = this.namesrvController.getRouteInfoManager().getHasUnitSubUnUnitTopicList();

    response.setBody(body);
    response.setCode(ResponseCode.SUCCESS);
    response.setRemark(null);
    return response;
  }

  private RemotingCommand updateConfig(ChannelHandlerContext ctx, RemotingCommand request) {
    log.info("updateConfig called by {}", RemotingHelper.parseChannelRemoteAddr(ctx.channel()));

    final RemotingCommand response = RemotingCommand.createResponseCommand(null);

    byte[] body = request.getBody();
    if (body != null) {
      String bodyStr;
      try {
        bodyStr = new String(body, MixAll.DEFAULT_CHARSET);
      } catch (UnsupportedEncodingException e) {
        log.error("updateConfig byte array to string error: ", e);
        response.setCode(ResponseCode.SYSTEM_ERROR);
        response.setRemark("UnsupportedEncodingException " + e);
        return response;
      }

      if (bodyStr == null) {
        log.error("updateConfig get null body!");
        response.setCode(ResponseCode.SYSTEM_ERROR);
        response.setRemark("string2Properties error");
        return response;
      }

      Properties properties = MixAll.string2Properties(bodyStr);
      if (properties == null) {
        log.error("updateConfig MixAll.string2Properties error {}", bodyStr);
        response.setCode(ResponseCode.SYSTEM_ERROR);
        response.setRemark("string2Properties error");
        return response;
      }

      this.namesrvController.getConfiguration().update(properties);
    }

    response.setCode(ResponseCode.SUCCESS);
    response.setRemark(null);
    return response;
  }

  private RemotingCommand getConfig(ChannelHandlerContext ctx, RemotingCommand request) {
    final RemotingCommand response = RemotingCommand.createResponseCommand(null);

    String content = this.namesrvController.getConfiguration().getAllConfigsFormatString();
    if (content != null && content.length() > 0) {
      try {
        response.setBody(content.getBytes(MixAll.DEFAULT_CHARSET));
      } catch (UnsupportedEncodingException e) {
        log.error("getConfig error, ", e);
        response.setCode(ResponseCode.SYSTEM_ERROR);
        response.setRemark("UnsupportedEncodingException " + e);
        return response;
      }
    }

    response.setCode(ResponseCode.SUCCESS);
    response.setRemark(null);
    return response;
  }
}
